package cn.lingyangwl.agile.demo.mq_sequence;

import cn.lingyangwl.agile.demo.mq_sequence.mapper.MqDataMapper;
import cn.lingyangwl.framework.core.utils.spring.SpringUtils;
import cn.lingyangwl.framework.tool.core.CollectionUtils;
import cn.lingyangwl.framework.tool.core.exception.BizException;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.springframework.context.event.EventListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * mq顺序消费
 * @author shenguangyang
 */
@Slf4j
@Service
public class MqDataServiceImpl extends ServiceImpl<MqDataMapper, MqData> implements MqDataService {
    @Resource
    private MqDataMapper mqDataMapper;

    /**
     * arg1: 业务编号
     * value: 消费记录
     */
    public static final String CONSUME_RECORD_KEY = "consume_record:%s";
    public static final String CONSUME_FAIL_KEY = "consume_fail:%s";
    public static final String CONSUME_LOCK_KEY = "consume_lock:%s";

    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private Redisson redisson;

    /**
     * 模拟消息队列中的存放的消息顺序
     */
    public static List<MqData> mqData = new ArrayList<>();

    @EventListener(classes = ConsumeFailEvent.class)
    public void listener(ConsumeFailEvent event) {
        log.info(event.getSource().toString());
        ConsumeFailMessage message = (ConsumeFailMessage) event.getSource();
        new Thread(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(50);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            processSequence(message, (data) -> {
                if (data.getSequenceNo() == 4) {
                    throw new BizException("测试连续失败");
                }
                System.out.println("消费消息-" + data.getMsgContent());
            });
        }).start();
    }

    @Override
    public void test() throws InterruptedException {
//        mqDataMapper.delete(Wrappers.<MqData>lambdaQuery().eq(MqData::getBizNo, "test"));
//        msgSequences = msgSequences.stream().sorted(Comparator.comparing(MsgSequence::getSequenceNo)).collect(Collectors.toList());
//        for (MqData mqData : MqDataServiceImpl.mqData) {
//            processSequence(mqData, (data) -> {
//                System.out.println("消费消息-" + data.getMsgContent());
//            });
//        }

        mqDataMapper.delete(Wrappers.<MqData>lambdaQuery().eq(MqData::getBizNo, "test"));
        System.out.println("start");
        long time = System.currentTimeMillis();
        for (MqData mqData : MqDataServiceImpl.mqData) {
            processSequence(mqData, (data) -> {
                if (data.getSequenceNo() == 4) {
                    throw new RuntimeException("测试失败");
                }
                System.out.println("消费消息-" + data.getMsgContent());
            });
        }
        System.out.println("end-time: " + (System.currentTimeMillis() - time) + " ms");
    }

    /**
     * 获取消费记录
     */
    public MqData getConsumeRecord(MqData mqData) {
        Object obj = redisTemplate.opsForValue().get(String.format(CONSUME_RECORD_KEY, mqData.getBizNo()));
        if (Objects.isNull(obj)) {
            return null;
        }
        return JSON.parseObject(String.valueOf(obj), MqData.class);
    }

    /**
     * 获取消费记录
     */
    public void addConsumeRecord(MqData mqData) {
        redisTemplate.opsForValue().set(String.format(CONSUME_RECORD_KEY, mqData.getBizNo()), JSON.toJSONString(mqData), 24, TimeUnit.HOURS);
    }

    public void handleWaitMessage(MqData consumeRecord, Consumer<MqData> consumer) {
        Integer sequenceNo = consumeRecord.getSequenceNo();
        IConsumeRule startConsumeRule = TestConsumeRule.getStartRule();

        Integer preSequenceNo = Objects.equals(sequenceNo, startConsumeRule.getCurrentNo()) ? sequenceNo : sequenceNo - 1;

        // 从数据库中查找消息
        LambdaQueryWrapper<MqData> lqw = MqData.lqw()
                .eq(MqData::getBizNo, consumeRecord.getBizNo())
                .ge(MqData::getSequenceNo, preSequenceNo)
                .orderByAsc(MqData::getSequenceNo);

        List<MqData> msgSequencesList = mqDataMapper.selectList(lqw).stream()
                .filter(e -> e.getSequenceNo().compareTo(preSequenceNo) >= 0).collect(Collectors.toList());
        if (CollectionUtils.isEmpty(msgSequencesList)) {
            return;
        }

        // 如果前一个消息没有被处理, 就直接返回
        if (!Objects.equals(sequenceNo, startConsumeRule.getCurrentNo())) {
            MqData preMqData = msgSequencesList.stream().filter(e -> e.getSequenceNo().compareTo(preSequenceNo) == 0)
                    .findFirst().orElse(null);
            if (preMqData == null || !preMqData.getStatus().equals(MqData.Status.PROCESSED.getCode())) {
                return;
            }
            //移除前一个已被处理的消息
            msgSequencesList.remove(preMqData);
        }

        // 开始处理消息
        List<MqData> processedList = new ArrayList<>();
        MqData lastProcessedMqData = null;
        ConsumeFailMessage consumeFailMessage = null;

//        IConsumeRule consumeRule = TestConsumeRule.ofByCurrentNo(sequenceNo);
        boolean isFail = false;
        for (MqData mqData : msgSequencesList) {
            sequenceNo++;
            // 如果消息被处理, 就进行下一次循环
            if (mqData.getStatus().equals(MqData.Status.PROCESSED.getCode())) {
                continue;
            }

            IConsumeRule preRule = TestConsumeRule.getPreRule(sequenceNo - 1);
            // 如果处理失败则跳过
            if (MqData.Status.PROCESS_FAIL.getCode().equals(mqData.getStatus())
                    && mqData.getErrorCount() > MqData.MAX_ERROR_COUNT) {
                continue;
            }

            if (preRule.getNextNoList().contains(mqData.getSequenceNo())) {
                try {
                    consumer.accept(mqData);
                    mqData.setStatus(MqData.Status.PROCESSED.getCode());
                } catch (Exception e) {
                    log.error("error: {}, data: {}", e.getMessage(), JSON.toJSONString(mqData));
                    mqData.setErrorCount(Optional.ofNullable(mqData.getErrorCount()).orElse(0) + 1);
                    mqData.setErrorContent(e.getMessage().substring(0, Math.min(e.getMessage().length(), 512)));
                    mqData.setStatus(MqData.Status.PROCESS_FAIL.getCode());
                    // 判断错误次数是否达到最大值
                    if (mqData.getErrorCount() <= MqData.MAX_ERROR_COUNT) {
                        consumeFailMessage = new ConsumeFailMessage();
                        BeanUtil.copyProperties(mqData, consumeFailMessage);
                        // 在缓存中记录一下
                        break;
                    }
                    // 只有达到最大失败次数才设置成失败
                    mqData.setStatus(MqData.Status.PROCESS_FAIL.getCode());
                }
                processedList.add(mqData);
                lastProcessedMqData = mqData;
            } else {
                break;
            }
        }

        // 记录被消费的最后一条消息
        if (Objects.nonNull(lastProcessedMqData) && lastProcessedMqData.getStatus().equals(MqData.Status.PROCESSED.getCode())) {
            addConsumeRecord(lastProcessedMqData);
        }

        // 更新处理的消息数据
        if (CollectionUtils.isNotEmpty(processedList)) {
            this.saveOrUpdateBatch(processedList);
        }

        // 判断是否有消费失败的消息
        if (Objects.nonNull(consumeFailMessage)) {
            SpringUtils.getApplicationContext().publishEvent(new ConsumeFailEvent(consumeFailMessage));
        }
    }

    @Override
    public void processSequence(MqData mqData, Consumer<MqData> consumer) {
        RLock rLock = redisson.getLock(String.format(CONSUME_LOCK_KEY, mqData.getBizNo()));
        rLock.lock(5, TimeUnit.SECONDS);
        try {
            mqData.setStatus(MqData.Status.WAIT_PROCESS.getCode());
            LambdaQueryWrapper<MqData> lqw = MqData.lqw()
                    .eq(MqData::getBizNo, mqData.getBizNo())
                    .eq(MqData::getStatus, MqData.Status.PROCESSED.getCode());

            long processedCount =  mqDataMapper.selectCount(lqw);
            if (processedCount ==  TestConsumeRule.values().length) {
                return;
            }

            Integer startSequenceNo = TestConsumeRule.getStartRule().getCurrentNo();
            if (startSequenceNo.compareTo(mqData.getSequenceNo()) == 0) {
                consumer.accept(mqData);
                // 将消息存入到数据库中
                mqData.setStatus(MqData.Status.PROCESSED.getCode());
                mqDataMapper.saveMsgData(mqData);

//                addConsumeRecord(mqData);
                handleWaitMessage(mqData, consumer);
                return;
            }

            mqDataMapper.saveMsgData(mqData);
            handleWaitMessage(mqData, consumer);
        } finally {
            if (rLock.isLocked() && rLock.isHeldByCurrentThread()) {
                rLock.unlock();
            }
        }
    }

    static {
        int dataLength = 1024;
        MqData msg01 = MqData.builder()
                .sequenceNo(1).bizNo("test").msgContent("test-1-" + RandomUtil.randomString(dataLength)).build();

        MqData msg04 = MqData.builder()
                .sequenceNo(4).bizNo("test").msgContent("test-4-" + RandomUtil.randomString(dataLength)).build();

        MqData msg02 = MqData.builder()
                .sequenceNo(2).bizNo("test").msgContent("test-2-" + RandomUtil.randomString(dataLength)).build();

        MqData msg03 = MqData.builder()
                .sequenceNo(3).bizNo("test").msgContent("test-3-" + RandomUtil.randomString(dataLength)).build();

        MqData msg06 = MqData.builder()
                .sequenceNo(6).bizNo("test").msgContent("test-6-" + RandomUtil.randomString(dataLength)).build();

        MqData msg05 = MqData.builder()
                .sequenceNo(5).bizNo("test").msgContent("test-5-" + RandomUtil.randomString(dataLength)).build();

        MqData msg07 = MqData.builder()
                .sequenceNo(7).bizNo("test").msgContent("test-7-" + RandomUtil.randomString(dataLength)).build();

        mqData.add(msg01);
//        mqData.add(msg05);
        mqData.add(msg02);
        mqData.add(msg04);
        mqData.add(msg03);
        mqData.add(msg06);
        mqData.add(msg07);

    }
}
